package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/7
 *
 *
 *  按照类型： 时间，计数
 *  按照特征:  滚动，滑动，会话
 *  按照计算方式: 全局，keyed窗口(非全局)
 *
 *
 *   滚动的，全局的，计数窗口(size = 3,slide = 3)
 *      每来3条数据，就算一次。
 *      窗口中最多存放3条
 *      不区分key
 *
 *   -----------------
 *   调用了窗口后，流会变为全局窗口流。
 *      AllWindowedStream.process(ProcessAllWindowFunction <IN, OUT, W extends Window>)
 *              IN:输入
 *              OUT:输出
 *              W： 窗口对象
 *                      TimeWindow: 是时间窗口
 *                      GlobalWindow: 是计数窗口
 *
 *  ------------------
 *      全局窗口： 指开窗后，下游窗口计算的并行度只能是1，是否设置都必须是1.
 *              是否keyBy，都会添加keyBy运算，保证运算是keyBy的运算。
 *
 *
 *
 *
 *
 *
 *
 */
public class Demo1_CountWindowAll
{
    public static void main(String[] args) {
        

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        AllWindowedStream<WaterSensor, GlobalWindow> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            //.keyBy(WaterSensor::getId)
            // 滚动
            //.countWindowAll(3);
            // 滑动 每来2条数据，就算一次，窗口中最多容纳3条
            .countWindowAll(3,2);

        ds
           .process(new ProcessAllWindowFunction<WaterSensor, String, GlobalWindow>()
           {
               /*
                    ProcessAllWindowFunction 中的 process 是窗口触发运算时，只执行一次！
                        不要和之前的搞混淆，之前的是来一条数据，就触发一次运算。

                        Context context： 编程上下文，从中获取工具，信息
                        Iterable<WaterSensor> elements： 可以迭代的集合，存储了窗口中所有的元素
                        Collector<String> out; 输出数据
                */
               @Override
               public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                   out.collect(MyUtil.parseToList(elements).toString());

               }
           }).setParallelism(1)
           .print();

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
